package com.sskj.mqtt.many;


import com.sskj.mqtt.many.mapper.MqttConfigProperties;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Component;

import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @Author: Like
 * @Date: 2025/3/3 15:40
 * @Description: 项目启动 连接emqx 订阅主题
 */
@Slf4j
@Component
public class MqttClientListener implements ApplicationListener<ContextRefreshedEvent> {


    /**
     * 是否已经初始化
     */
    private volatile AtomicBoolean isInit = new AtomicBoolean(false);

    @Autowired
    private MqttConfigProperties mqttConfig;

    /**
     * 项目启动 监听主题
     *
     * @param event
     */
    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        if (!isInit.compareAndSet(false, true)) {
            return;
        }

        mqttConfig.getClients().forEach(clientConfig -> {
            try {
                MqttClientConnect client = createMqttClient(clientConfig);
                registerClient(clientConfig, client);
            } catch (MqttException e) {
                log.error("MQTT客户端初始化失败: {}", clientConfig.getClientId(), e);
            }
        });
    }

    /**
     * 创建MQTT客户端
     *
     * @param config
     * @return
     * @throws MqttException
     */
    private MqttClientConnect createMqttClient(MqttConfigProperties.MqttClientConfig config) throws MqttException {
        MqttClientConnect client = new MqttClientConnect();
        client.setMqttClientId(config.getClientId());

        String serverUri = buildServerUri(config);
        MqttClientCallback callback = new MqttClientCallback(config.getClientId());

        client.setMqttClient(
                serverUri,
                config.getClientId(),
                config.getUsername(),
                config.getPassword(),
                config.isCleanSession(),
                config.getConnectionTimeout(),
                config.getKeepAliveInterval(),
                callback
        );

        subscribeTopics(client, MqttClientMap.getTopicsByClientId(config.getClientId()));
        return client;
    }

    /**
     * 构建MQTT服务器地址
     *
     * @param config
     * @return
     */
    private String buildServerUri(MqttConfigProperties.MqttClientConfig config) {
        // 如果配置中serverHost已包含协议头，则直接使用
        if (config.getServerHost().startsWith("tcp://") ||
                config.getServerHost().startsWith("ssl://")) {
            return config.getServerHost() + ":" + config.getServerPort();
        }
        return "tcp://" + config.getServerHost() + ":" + config.getServerPort();
    }

    /**
     * 订阅主题
     *
     * @param client
     * @param topics
     */
    private void subscribeTopics(MqttClientConnect client, String[] topics) {
        if (topics.length != 0) {
            try {
                client.sub(topics);
                log.info("客户端 [{}] 订阅主题: {}", client.getMqttClientId(), topics);
            } catch (MqttException e) {
                log.error("订阅主题失败: {}", topics, e);
            }
        }
    }

    /**
     * 注册客户端
     *
     * @param config
     * @param client
     */
    private void registerClient(MqttConfigProperties.MqttClientConfig config, MqttClientConnect client) {
        MqttClientConnect.mqttClients.put(config.getClientId(), client);
        log.info("MQTT客户端 [{}] 初始化完成 - 连接地址: {}:{}",
                config.getClientId(),
                config.getServerHost(),
                config.getServerPort());
    }

}
